package com.gxlevi.batch_process.task

import com.gxlevi.batch_process.bean.OrderRecordWide
import com.gxlevi.batch_process.util.HBaseUtil
import org.apache.flink.api.scala._

case class PayMethodMoneyCount(
                                var payMethod: String, //支付方式
                                var date: String, //日期
                                var moneyCount: Double, //订单金额
                                var count: Long //订单总数
                              )

object PayMethodMoneyCountTask {
  def process(orderRecordDataSet: DataSet[OrderRecordWide]) = {
    val payMethodDataSet = orderRecordDataSet.flatMap {
      orderRecordWide =>
        List(
          PayMethodMoneyCount(orderRecordWide.payMethod, orderRecordWide.yearMonthDay, orderRecordWide.payAmount.toDouble, 1),
          PayMethodMoneyCount(orderRecordWide.payMethod, orderRecordWide.yearMonth, orderRecordWide.payAmount.toDouble, 1),
          PayMethodMoneyCount(orderRecordWide.payMethod, orderRecordWide.year, orderRecordWide.payAmount.toDouble, 1)
        )
    }

    val groupDataSet = payMethodDataSet.groupBy {
      payMethodMoneyCount => payMethodMoneyCount.payMethod + payMethodMoneyCount.date
    }

    val reduceDataSet = groupDataSet.reduce {
      (p1, p2) => {
        PayMethodMoneyCount(p1.payMethod, p1.date, p1.moneyCount + p2.moneyCount, p1.count + p2.count)
      }
    }
    //落地
    reduceDataSet.collect().foreach {
      payMethodMoneyCount =>
        val tableName = "analysis_payment"
        val clfName = "info"
        val rowKey = payMethodMoneyCount.payMethod + ":" + payMethodMoneyCount.date

        var payMethodColumn = "payMethod"
        var dateColumn = "date"
        var moneyCountColumn = "moneyCount"
        var countColumn = "count"

        HBaseUtil.putMapData(tableName, rowKey, clfName, Map(
          payMethodColumn -> payMethodMoneyCount.payMethod,
          dateColumn -> payMethodMoneyCount.date,
          moneyCountColumn -> payMethodMoneyCount.moneyCount,
          countColumn -> payMethodMoneyCount.count
        ))
    }
  }
}
